package com.gitee.andy.zpush.node;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.gitee.andy.zpush.kernel.SessionContext;
import com.gitee.andy.zpush.kernel.TcpServer;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.util.concurrent.DefaultThreadFactory;

public class ZpushNodeMain {

	//private static final Logger log = LoggerFactory.getLogger(ZpushNodeMain.class);

	public static void main(String[] args) {

		ConcurrentLinkedQueue<Object> receiveMsgQueue = new ConcurrentLinkedQueue<>();

		ConcurrentLinkedQueue<Object> pushMsgQueue = new ConcurrentLinkedQueue<>();

		ScheduledExecutorService ses = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors() * 2,
				new DefaultThreadFactory("write2client-pool"));

		SessionContext sessionContext = SessionContext.create(ses);

		SessionContext session4CenterContext = SessionContext.createRand(ses);

		TcpServer tcpServer = new TcpServer("node-bossGroup", "node- workerGroup");
		tcpServer.bind(8007, new ChannelInitializer<SocketChannel>() {

			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(
						new ObjectDecoder(1024 * 1024 * 8,
								ClassResolvers.softCachingConcurrentResolver(this.getClass().getClassLoader())),
						new ObjectEncoder(), new EchoServerHandler(sessionContext, receiveMsgQueue));
			}
		});
		
		tcpServer.bind(8008, new ChannelInitializer<SocketChannel>() {

			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(
						new ObjectDecoder(1024 * 1024 * 8,
								ClassResolvers.softCachingConcurrentResolver(this.getClass().getClassLoader())),
						new ObjectEncoder(), new Push2CenterHandler(session4CenterContext, pushMsgQueue));
			}
		});

		ses.scheduleWithFixedDelay(() -> {
			Object obj;
			while ((obj = pushMsgQueue.poll()) != null) {
				//log.info("poll msg:{}", obj);
				sessionContext.write(obj);
			}
		}, 50, 200, TimeUnit.MILLISECONDS);
		
		ses.scheduleWithFixedDelay(() -> {
			Object obj;
			while ((obj = receiveMsgQueue.poll()) != null) {
				//log.info("poll msg:{}", obj);
				session4CenterContext.write(obj);
			}
		}, 50, 200, TimeUnit.MILLISECONDS);

		Runtime.getRuntime().addShutdownHook(new Thread(() -> {
			sessionContext.shutdown();
			tcpServer.shutdown();
		}));
	}
}
